package _01_quick_start._8_2_4_broadcasting_a_stream

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._

object Main extends App {

  implicit val system       = ActorSystem("QuickStart")
  implicit val materializer = ActorMaterializer()
  implicit val ec           = system.dispatcher

  final case class Author(handle: String)

  final case class Hashtag(name: String)

  final case class Tweet(author: Author, timestamp: Long, body: String) {
    def hashtags: Set[Hashtag] =
      body.split(" ").collect { case t if t.startsWith("#") => Hashtag(t) }.toSet
  }

  val akkaTag = Hashtag("#akka")

  /*
   * 非结构化`streaming`处理
   */

  // source
  val tweets: Source[Tweet, NotUsed] = Source(
    Set(
      Tweet(Author("Martin"), System.currentTimeMillis(), "Ordersky #akka creator"),
      Tweet(Author("Barudisu"), System.currentTimeMillis(), "#akka leaner"),
      Tweet(Author("Petter"), System.currentTimeMillis(), "Mahina #scala practise")
    ))

  // 广播(broadcasting a stream)
  val writeAuthors  = Sink foreach println
  val writeHashtags = Sink foreach println

  val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit b =>
    import GraphDSL.Implicits._

    val bcast = b.add(Broadcast[Tweet](2))
    tweets ~> bcast.in
    bcast.out(0) ~> Flow[Tweet].filter(_.hashtags.contains(akkaTag)).map(_.author) ~> writeAuthors
    bcast.out(1) ~> Flow[Tweet].mapConcat(_.hashtags.toList) ~> writeHashtags
    ClosedShape
  })

  val done = g.run()
}
